

package com.hazelcast.client.impl.protocol.codec.builtin;

import com.hazelcast.client.impl.protocol.ClientMessage;
import com.hazelcast.client.impl.protocol.codec.holder.VectorPairHolder;
import com.hazelcast.vector.VectorValues;

import java.util.ArrayList;
import java.util.Collection;
import java.util.List;
import java.util.function.BiConsumer;
import java.util.function.Function;
import java.util.stream.Collectors;

import static com.hazelcast.client.impl.protocol.ClientMessage.*;
import static com.hazelcast.client.impl.protocol.codec.builtin.CodecUtil.nextFrameIsDataStructureEndFrame;
import static com.hazelcast.client.impl.protocol.codec.builtin.CodecUtil.nextFrameIsNullEndFrame;

public final class ListMultiFrameCodec {

    private ListMultiFrameCodec() {
    }

    public static <T> void encode(ClientMessage clientMessage, Iterable<T> collection, BiConsumer<ClientMessage, T> encodeFunction) {
        clientMessage.add(BEGIN_FRAME.copy());
        for (T item : collection) {
            encodeFunction.accept(clientMessage, item);
        }
        clientMessage.add(END_FRAME.copy());
    }

    public static <T> void encodeContainsNullable(ClientMessage clientMessage, Iterable<T> collection, BiConsumer<ClientMessage, T> encodeFunction) {
        clientMessage.add(BEGIN_FRAME.copy());
        for (T item : collection) {
            if (item == null) {
                clientMessage.add(NULL_FRAME.copy());
            } else {
                encodeFunction.accept(clientMessage, item);
            }
        }
        clientMessage.add(END_FRAME.copy());
    }

    public static <T> void encodeNullable(ClientMessage clientMessage, Collection<T> collection, BiConsumer<ClientMessage, T> encodeFunction) {
        if (collection == null) {
            clientMessage.add(NULL_FRAME.copy());
        } else {
            encode(clientMessage, collection, encodeFunction);
        }
    }

    public static <T> List<T> decode(ClientMessage.ForwardFrameIterator iterator, Function<ClientMessage.ForwardFrameIterator, T> decodeFunction) {
        List<T> result = new ArrayList<>();
        //begin frame, list
        iterator.next();
        while (!nextFrameIsDataStructureEndFrame(iterator)) {
            result.add(decodeFunction.apply(iterator));
        }
        //end frame, list
        iterator.next();
        return result;
    }

    public static <T> List<T> decodeContainsNullable(ClientMessage.ForwardFrameIterator iterator, Function<ClientMessage.ForwardFrameIterator, T> decodeFunction) {
        List<T> result = new ArrayList<>();
        //begin frame, list
        iterator.next();
        while (!nextFrameIsDataStructureEndFrame(iterator)) {
            result.add(nextFrameIsNullEndFrame(iterator) ? null : decodeFunction.apply(iterator));
        }
        //end frame, list
        iterator.next();
        return result;
    }

    public static <T> List<T> decodeNullable(ClientMessage.ForwardFrameIterator iterator, Function<ClientMessage.ForwardFrameIterator, T> decodeFunction) {
        return nextFrameIsNullEndFrame(iterator) ? null : decode(iterator, decodeFunction);
    }

    // specialized implementations for VectorValues
    public static void encode(ClientMessage clientMessage, VectorValues vectors, BiConsumer<ClientMessage, VectorPairHolder> encodeFunction) {
        if (vectors instanceof VectorValues.SingleVectorValues svv) {
            encode(clientMessage, List.of(new VectorPairHolder(VectorPairHolder.SINGLE_VECTOR_NAME, VectorPairHolder.DENSE_FLOAT_VECTOR, svv.vector())), encodeFunction);
        } else
            if (vectors instanceof VectorValues.MultiIndexVectorValues mvv) {
                var holders = mvv.indexNameToVector().entrySet().stream().map(entry -> new VectorPairHolder(entry.getKey(), VectorPairHolder.DENSE_FLOAT_VECTOR, entry.getValue())).collect(Collectors.toList());
                encode(clientMessage, holders, encodeFunction);
            }
    }

    public static <T> void encodeNullable(ClientMessage clientMessage, VectorValues collection, BiConsumer<ClientMessage, VectorPairHolder> encodeFunction) {
        if (collection == null) {
            clientMessage.add(NULL_FRAME.copy());
        } else {
            encode(clientMessage, collection, encodeFunction);
        }
    }
}
